package com.yanwq.nwa.im;

import android.util.Log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;

/**
 * Created by yanweiqiang on 2017/2/24 0024.
 */
public class MessageMgr {
    private static MessageMgr instance;

    synchronized static private void syncInit() {
        if (instance == null) {
            instance = new MessageMgr();
        }
    }

    static public MessageMgr getInstance() {
        syncInit();
        return instance;
    }

    private final String TAG = "MessageMgr";
    private final BlockingQueue<String> queue;
    private Channel channel;
    private Map<String, Callback> callbackMap;

    private MessageMgr() {
        int capacity = 10;
        queue = new ArrayBlockingQueue<String>(capacity);
        channel = null;
        callbackMap = new HashMap<>();
    }

    public void initBlockingMessageQueue(final Channel channel) throws InterruptedException {
        this.channel = channel;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                WebSocketFrame frame = new TextWebSocketFrame(queue.take());
                this.channel.writeAndFlush(frame);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        Log.i(TAG, "thread:" + Thread.currentThread().getName() + "old channel closed.");
    }

    public void callSuccess(String data) {
        JSONObject jsonObject = JSON.parseObject(data);

        if (!jsonObject.containsKey("msg_id")) {
            return;
        }

        Callback callback = callbackMap.get(jsonObject.getString("msg_id"));

        if (callback == null) {
            return;
        }

        callback.call(data);
    }

    public void send(String msg, Callback callback) {
        JSONObject jsonObject = JSON.parseObject(msg);
        callbackMap.put(jsonObject.getString("msg_id"), callback);
        Log.d("MessageMgr", "send<->" + msg);
        if (Connector.getInstance().getStat() != Connector.ConnectionStat.CONNECTED) {
            EventMgr.getInstance().triggerFailure(EventCallback.Failure.DISCONNECTED);
            return;
        }
        queue.add(msg);
    }

    public void sendPing() {
        if (channel == null) {
            Log.i("MessageMgr", "channel is null");
            return;
        }

        if (!channel.isActive()) {
            Log.i("MessageMgr", "channel is inActive");
            return;
        }

        channel.writeAndFlush(new PingWebSocketFrame(Unpooled.copiedBuffer(new byte[]{8, 1, 8, 1})));
    }

    public void sendClose() {
        channel.writeAndFlush(new CloseWebSocketFrame());
        try {
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public interface Callback {
        void onSuccess(String data);

        void onFailure();
    }
}
